package com.zhang.third.day03;

import com.zhang.third.utils.ClickEventSource;
import com.zhang.third.utils.Event;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @title: 自定义幂等写入MySQL
 * @author: zhang
 * @date: 2022/4/6 09:36
 */
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickEventSource())
                .addSink(new MyJDBC());

        env.execute();
    }

    public static class MyJDBC extends RichSinkFunction<Event> {
        private Connection connection;
        private PreparedStatement insertStmt;
        private PreparedStatement updateStmt;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection(
                    "jdbc:mysql://localhost:3306/userbehavior?useSSl=false",
                    "root",
                    "000000"
            );

            insertStmt = connection.prepareStatement("INSERT INTO clicks (username,url) VALUES (?,?)");
            updateStmt = connection.prepareStatement("UPDATE clicks SET url = ? WHERE username = ?");
        }

        // 数据每来一条执行一次
        @Override
        public void invoke(Event value, Context context) throws Exception {
            updateStmt.setString(1, value.value);
            updateStmt.setString(2, value.key);
            updateStmt.execute();

            if (updateStmt.getUpdateCount() == 0) {
                insertStmt.setString(1, value.key);
                insertStmt.setString(2, value.value);
                insertStmt.execute();
            }
        }

        @Override
        public void close() throws Exception {
            insertStmt.close();
            updateStmt.close();
            connection.close();
        }
    }
}
